package com.frinder.demo4;

import akka.actor.*;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.routing.ActorRefRoutee;
import akka.routing.RoundRobinRoutingLogic;
import akka.routing.Routee;
import akka.routing.Router;
import com.frinder.demo3.InboxTest;
import com.google.common.collect.Lists;
import com.typesafe.config.ConfigFactory;

import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * Created by frinder on 2017/7/24.
 */
public class RouterTest extends UntypedActor {

    final LoggingAdapter logger = Logging.getLogger(getContext().system(), this);

    public Router router;

    {
        List<Routee> routees = Lists.newArrayList();
        for (int i = 1; i <= 5; i++) {
            ActorRef worker = getContext().actorOf(Props.create(InboxTest.class), "worker_" + i);
            getContext().watch(worker);
            routees.add(new ActorRefRoutee(worker));
        }
        /**
         * RoundRobinRoutingLogic: 轮询
         * BroadcastRoutingLogic: 广播
         * RandomRoutingLogic: 随机
         * SmallestMailboxRoutingLogic: 空闲
         */
        router = new Router(new RoundRobinRoutingLogic(), routees);
    }

    static AtomicBoolean isRunning = new AtomicBoolean(true);

    @Override
    public void onReceive(Object o) throws Exception {
        if (o instanceof InboxTest.Msg) {
            logger.info("{}", o);
            router.route(o, getSender());
        } else if (o instanceof Terminated) {
            ActorRef ref = ((Terminated) o).actor();
            router = router.removeRoutee(ref);
            logger.warning("{} has been deleted! router size {}.", ref.path(), router.routees().size());
            if (router.routees().isEmpty()) {
                logger.warning("no available actor, shutdown system.");
                isRunning.compareAndSet(true, false);
                getContext().system().terminate();
            }
        } else {
            unhandled(o);
        }
    }

    public static void main(String[] args) throws Exception {
        ActorSystem system = ActorSystem.create("router", ConfigFactory.load("akka.config"));
        ActorRef ref = system.actorOf(Props.create(RouterTest.class), "router");
        int i = 1;
        while (isRunning.get()) {
            ref.tell(InboxTest.Msg.WORKING, ActorRef.noSender());
            if (i % 10 == 0) {
                ref.tell(InboxTest.Msg.CLOSE, ActorRef.noSender());
            }
            Thread.sleep(500);
            i++;
        }
    }
}
